using System;
using System.IO;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using Microsoft.BizTalk.TransportProxy.Interop;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.XLANGs.BaseTypes;

namespace Microsoft.Samples.BizTalk.Adapter.Tcp.Receive
{
    /// <summary>
    ///     This class represents a worker thread that is used to dequeue work from pending sockets
    ///     and submit in batches to the messaging engine.
    ///     Each worker thread has its own queue (FIFO) that a dispatcher uses to insert pending sockets.
    /// </summary>
    /// <author>
    ///     Paul Brown
    /// </author>
    public class BatchWorkerThread : WorkerThread
    {
        #region Private Instance Members

        private ReceiveAdapter  _receiveAdapter;    // The receive adapter

        #endregion

        #region Constructor

        /// <summary>
        ///     This constructor initializes the instance.
        /// </summary>
        public BatchWorkerThread(IAdapter adapter)
            : base(adapter)
        {
            // Get thread name prefix
            this.ThreadNamePrefix = TraceHelper.ResourceManager.GetString(ResourceConstants.ReceiveBatchWorkerThreadName);

            // We know the adapter will be a receive adapter, store a reference to it
            this._receiveAdapter = (ReceiveAdapter)adapter;
        }

        #endregion

        #region Public Instance Properties

        /// <summary>
        ///     This property returns a reference to the receive adapter.
        /// </summary>
        public ReceiveAdapter ReceiveAdapter
        {
            get { return (this._receiveAdapter); }
        }

        #endregion

        #region Private Instance Methods

        /// <summary>
        ///     This method creates a BizTalk message for submission into the batch.
        /// </summary>
        private IBaseMessage CreateMessage(Stream stream, IBaseMessageFactory messageFactory, BatchWorkItem workItem)
        {
            return (ReceiveLocation.CreateMessage(stream, messageFactory, workItem.ReceiveLocation));
        }

        /// <summary>
        ///     This method gets the data from the socket and closes the socket if it is a non-duplex receive location.
        ///     It has the potential for returning multiple streams depending on how many messages are stacked on the socket.
        /// </summary>
        private Stream[] GetData(BatchWorkItem workItem, out bool isLastStreamNotComplete)
        {
            // Receive the data
            bool lastStreamNotComplete;
            Stream[] streams = SocketIO.ReceiveData(this.Adapter,
                                                    workItem.ClientSocket,
                                                    workItem.ReceiveLocation.FrameStart,
                                                    workItem.ReceiveLocation.FrameEnd,
                                                    workItem.ReceiveLocation.DataEncoding,
                                                    workItem.ReceiveLocation.ReceiveDataBufferSize,
                                                    workItem.ReceiveLocation.AcceptMultipleMessages,
                                                    out lastStreamNotComplete);

            // Close the socket if it is a non-duplex one-way socket
            if ((!workItem.ReceiveLocation.IsDuplex) && (!workItem.ReceiveLocation.IsTwoWay))
            {
                if (workItem.ClientSocket != null)
                {
                    if (workItem.ClientSocket.Connected)
                    {
                        workItem.ClientSocket.Shutdown(SocketShutdown.Receive);
                        workItem.ClientSocket.Close(workItem.ReceiveLocation.MainCloseTimeout);
                    }
                }
            }

            // Return array
            isLastStreamNotComplete = lastStreamNotComplete;
            return (streams);
        }

        #endregion

        #region Abstract WorkerThread Members

        #region Public Instance Properties

        /// <summary>
        ///     This property returns the type of a BatchWorkItem which is compatible with this worker thread.
        /// </summary>
        public override Type WorkItemType
        {
            get { return (typeof(BatchWorkItem)); }
        }

        #endregion

        #region Protected Instance Methods

        /// <summary>
        ///     This method is the delegate that runs the main worker thread for unordered batches.
        /// /// 
        /// 
        /// found one thing slightly dodgy in code - the batch is created before the while loop starts in BatchWorkerThread.cs.  
        /// If the threads are busy then it will never sleep.  However, if it submits a batch then it gets reused and who knows 
        /// what state the new batch is in...
        /// 
        /// This will only happen it seems under stress because generally the worker thread will sleep or another one is used
        /// Once a batch has been submitted, it needs to create a new fresh one. 
        /// Even if it is still in the while loop.
        /// 
        /// But it doesn't want to do it every iteration, only when the previous batch has been submitted.
        /// </summary>
        protected override void Thread()
        {
            // variable used to identify if batch was submitted
            bool submitted = false;

            // Variables used to control batching
            bool            threadInterrupted = false;
            int             messageCount;
            long            batchDataSize;
            Stopwatch       timer = new Stopwatch();
            BatchWorkItem   workItem = null;

            try
            {
                // Loop until we are interrupted
                while (true)
                {
                    // If we have work, start batching process
                    if (this.CurrentWorkLoad > 0)
                    {
                        if (this.Adapter.TraceSwitch.TraceVerbose)
                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveGetBatch, this.Name);

                        // Create a new batch for BizTalk
                        Batch batch = new Batch(this.Adapter);
                        batch.RetryCount = this.ReceiveAdapter.BatchRetryCount;
                        batch.RetryInterval = this.ReceiveAdapter.BatchRetryInterval;
                        batch.OrderedBatch = false;

                        // Reset batch controlling properties
                        messageCount = 0;
                        batchDataSize = 0;
                        timer.Reset();
                        timer.Start();

                        #region Loop while we have work items
                        // Loop while we have work items
                        while (this.CurrentWorkLoad > 0)
                        {
                            if (submitted == true)
                            {
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveInitialiseNewBatch, this.Name);
                                batch = new Batch(this.Adapter);
                                batch.RetryCount = this.ReceiveAdapter.BatchRetryCount;
                                batch.RetryInterval = this.ReceiveAdapter.BatchRetryInterval;
                                batch.OrderedBatch = false;

                                // Reset batch controlling properties
                                messageCount = 0;
                                batchDataSize = 0;
                                timer.Reset();
                                timer.Start();
                                submitted = false;
                            }
                            
                            // Dequeue item
                            workItem = (BatchWorkItem)this.DequeueWork();

                            if (this.Adapter.TraceSwitch.TraceVerbose)
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveDequeuedWork, workItem.ReceiveLocation.Uri, this.Name);

                            // Set its status
                            workItem.Status = WorkItemStatusTypes.InProgress;

                            // Get data
                            bool isLastMessageNotComplete;
                            workItem.DataStreams = this.GetData(workItem, out isLastMessageNotComplete);

                            bool lastMessage = false;
                            for (int i = 0; i < workItem.DataStreams.Length; i++)
                            {
                                // Get stream
                                Stream dataStream = workItem.DataStreams[i];

                                // Is it the last message in the array?
                                if (i == (workItem.DataStreams.Length - 1))
                                {
                                    lastMessage = true;
                                }

                                if (dataStream != null)
                                {
                                    // Create a BizTalk message for the work item
                                    IBaseMessage message = this.CreateMessage(dataStream, this.Adapter.TransportProxy.GetMessageFactory(), workItem);

                                    bool addedSuccessfully = false;
                                    while (!addedSuccessfully)
                                    {
                                        try
                                        {
                                            // Add to the batch
                                            if (workItem.ReceiveLocation.IsTwoWay)
                                            {
                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmitRequestMessageToBatch, workItem.ReceiveLocation.Uri);

                                                // Set up response timeout value
                                                DateTime expirationTime = DateTime.Now.AddMinutes(workItem.ReceiveLocation.ResponseMessageTimeout);
                                                string correlationToken = Guid.NewGuid().ToString();

                                                // Submit to batch
                                                if (lastMessage && isLastMessageNotComplete)
                                                {
                                                    // Suspend message
                                                    BatchMessage batchMessage = new BatchMessage(message, new Exception(TraceHelper.FormatEntry(ResourceConstants.SocketSuspendedPartialMessage, dataStream.Length)));
                                                    batch.MoveToSuspendQ(batchMessage);
                                                }
                                                else
                                                {
                                                    // Add a transmitter to wait for the response
                                                    Transmitter transmitter = new Transmitter(this.Adapter, workItem.ReceiveLocation, workItem.ClientSocket);
                                                    Transmitter.AddHandler(transmitter);

                                                    // Submit message
                                                    batch.SubmitRequestMessage(new BatchMessage(message, correlationToken, expirationTime, transmitter));
                                                }
                                            }
                                            else
                                            {
                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmitMessageToBatch, workItem.ReceiveLocation.Uri);
                                                
                                                // Submit to batch
                                                if (lastMessage && isLastMessageNotComplete)
                                                {
                                                    // Suspend message
                                                    BatchMessage batchMessage = new BatchMessage(message, new Exception(TraceHelper.FormatEntry(ResourceConstants.SocketSuspendedPartialMessage, dataStream.Length)));
                                                    batch.MoveToSuspendQ(batchMessage);
                                                }
                                                else
                                                {
                                                    // Submit message
                                                    batch.SubmitMessage(new BatchMessage(message));
                                                }
                                            }

                                            addedSuccessfully = true;
                                        }
                                        catch (ThreadInterruptedException)
                                        {
                                            // Asked to shut down, set flag and wait until this batch is done
                                            threadInterrupted = true;
                                        }
                                        catch (Exception e)
                                        {
                                            TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedAddingToBatch, workItem.ReceiveLocation.Uri, batch.BatchCookie, e.Message);
                                        }
                                    }

                                    // Increment controlling properties
                                    messageCount++;
                                    batchDataSize += dataStream.Length;
                                }
                            }
                        #endregion

                            // Anything in batch?
                            if (messageCount > 0)
                            {
                                bool readyToSubmit = false;

                                // Have we exceeded our batch limits?
                                if (messageCount >= this.ReceiveAdapter.BatchSize)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveReachedBatchSizeLimit, this.ReceiveAdapter.BatchSize, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (batchDataSize >= this.ReceiveAdapter.MaxBatchData)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveExceededMaxBatchDataLimit, this.ReceiveAdapter.MaxBatchData, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (timer.ElapsedMilliseconds >= this.ReceiveAdapter.BatchTimeout)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveExceededBatchTimeoutLimit, this.ReceiveAdapter.BatchTimeout, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (this.CurrentWorkLoad == 0)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmittingBatch, messageCount);

                                    readyToSubmit = true;
                                }

                                // Are we ready to submit?
                                if (readyToSubmit)
                                {
                                    // Add to array
                                    if (Batch.GetBatch(batch.BatchCookie) == null)
                                    {
                                        Batch.AddBatch(batch);
                                    }

                                    // Submit batch if limits exceeded or idle
                                    
                                    for (int i = 0; i < this.ReceiveAdapter.BatchRetryCount; i++)
                                    {
                                        try
                                        {
                                            // Submit batch
                                            batch.SubmitBatch();

                                            // Break out of loop, successful
                                            submitted = true;
                                            break;
                                        }
                                        catch (ThreadInterruptedException)
                                        {
                                            // Asked to shut down, set flag and wait until this batch is done
                                            threadInterrupted = true;
                                        }
                                        catch (Exception e)
                                        {
                                            // Trace
                                            TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedSubmittingBatchRetrying, batch.BatchCookie, e.Message);

                                            // Wait interval
                                            System.Threading.Thread.Sleep(this.ReceiveAdapter.BatchRetryInterval * 1000);
                                        }
                                    }

                                    // If not successful and exhausted all retries
                                    if (!submitted)
                                    {
                                        TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedSubmittingBatch, batch.BatchCookie);

                                        // Save data
                                        workItem.Save();
                                    }
                                }
                            }
                            else
                            {
                                // No messages in batch!
                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveNoMessagesInBatch);
                            }

                            // Update work item status
                            workItem.Status = WorkItemStatusTypes.Complete;
                            workItem = null;
                        }
                    }

                    // Stop timer
                    timer.Stop();

                    // Was the thread interrupted?
                    if (threadInterrupted)
                    {
                        // Throw error again to end thread
                        throw (new ThreadInterruptedException());
                    }

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadWaitingForWork, this.Name);

                    // Wait for signal that work has arrived
                    this.WaitForWork();
                }
            }
            catch (ThreadInterruptedException)
            {
                // Asked to shut down
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadInterrupted, this.Name);

                // If we have any in-flight work still, save it off
                if (workItem != null)
                {
                    workItem.Save();
                }

                while (this.CurrentWorkLoad > 0)
                {
                    workItem = (BatchWorkItem)this.DequeueWork();
                    workItem.Save();
                }
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.WorkerThreadGeneralError, this.Name, e.Message);

                // If we have an outstanding work item, push it back onto queue before restarting
                if (workItem != null)
                {
                    this.PushWorkOnQueue(workItem);
                }
               
            }
            finally
            {
                // Fire terminate event
                this.OnThreadTerminated(new EventArgs());
            }
        }

        #endregion

        #endregion
    }
}
